const amqp = require('amqplib');
const { promisify } = require('util');

// RabbitMQ 连接配置
const rabbitConfig = {
    protocol: 'amqp',
    hostname: 'localhost',
    port: 5672,
    username: 'admin',
    password: '123456',
    vhost: '/',
    heartbeat: 60,
    retryDelay: 3000 // 重连延迟
};

// 核心配置
const BASE_QUEUE_NAME = 'demo';
const NORMAL_QUEUE = BASE_QUEUE_NAME;
const DEAD_QUEUE = `${BASE_QUEUE_NAME}_dead`;
const NORMAL_ROUTING_KEY = NORMAL_QUEUE;
const DEAD_ROUTING_KEY = DEAD_QUEUE;
const EXCHANGE_NAME = `${BASE_QUEUE_NAME}_exchange`;
const DEAD_EXCHANGE_NAME = `${BASE_QUEUE_NAME}_dead_exchange`;

// 全局连接与通道管理（避免重复创建）
let connection = null;
let channel = null;

/**
 * 安全获取连接（自动重连）
 */
async function getConnection() {
    if (connection && connection.connection && connection.connection.state === 'open') {
        return connection;
    }

    try {
        connection = await amqp.connect(rabbitConfig);
        console.log('RabbitMQ 连接成功');

        // 监听连接关闭事件，自动重连
        connection.on('close', (err) => {
            console.error('连接已关闭，准备重连:', err.message);
            connection = null;
            channel = null;
            setTimeout(getConnection, rabbitConfig.retryDelay);
        });

        // 监听连接错误事件
        connection.on('error', (err) => {
            console.error('连接错误:', err.message);
        });

        return connection;
    } catch (error) {
        console.error('连接失败，将重试:', error.message);
        throw error; // 抛出错误让调用方处理
    }
}

/**
 * 安全获取通道（基于活跃连接）
 */
async function getChannel() {
    if (channel && channel.channel && channel.channel.state === 'open') {
        return channel;
    }

    const conn = await getConnection();
    channel = await conn.createChannel();

    // 监听通道错误
    channel.on('error', (err) => {
        console.error('通道错误:', err.message);
        channel = null;
    });

    return channel;
}

/**
 * 初始化 RabbitMQ 环境（确保幂等性）
 */
async function initRabbitMQ() {
    try {
        const ch = await getChannel();

        // 声明交换机和队列（使用 assert 确保幂等性，重复调用不会报错）
        await Promise.all([
            ch.assertExchange(EXCHANGE_NAME, 'direct', { durable: true, autoDelete: false }),
            ch.assertExchange(DEAD_EXCHANGE_NAME, 'direct', { durable: true, autoDelete: false }),
            ch.assertQueue(NORMAL_QUEUE, {
                durable: true,
                autoDelete: false,
                deadLetterExchange: DEAD_EXCHANGE_NAME,
                deadLetterRoutingKey: DEAD_ROUTING_KEY,
                maxLength: 1000
            }),
            ch.assertQueue(DEAD_QUEUE, { durable: true, autoDelete: false })
        ]);

        // 绑定队列与交换机
        await Promise.all([
            ch.bindQueue(NORMAL_QUEUE, EXCHANGE_NAME, NORMAL_ROUTING_KEY),
            ch.bindQueue(DEAD_QUEUE, DEAD_EXCHANGE_NAME, DEAD_ROUTING_KEY)
        ]);

        console.log(`环境初始化完成:
      正常队列: ${NORMAL_QUEUE} → 死信队列: ${DEAD_QUEUE}
    `);
    } catch (error) {
        console.error('初始化失败:', error.message);
        throw error; // 允许上层处理
    }
}

/**
 * 生产者：发送消息（使用共享通道）
 */
async function produceMessage(content, ttl = 10000) {
    try {
        const ch = await getChannel();
        const result = ch.publish(
            EXCHANGE_NAME,
            NORMAL_ROUTING_KEY,
            Buffer.from(content),
            { persistent: true, expiration: ttl.toString(), timestamp: Date.now() }
        );

        if (result) {
            console.log(`[生产者] 发送成功: ${content}（过期时间: ${ttl}ms）`);
        } else {
            console.warn(`[生产者] 消息可能未发送成功（通道忙）: ${content}`);
        }
    } catch (error) {
        console.error('[生产者] 发送失败:', error.message);
        // 可添加消息重试逻辑
    }
}

/**
 * 正常队列消费者
 */
async function consumeNormalQueue() {
    try {
        const ch = await getChannel();
        console.log(`[${NORMAL_QUEUE} 消费者] 开始监听...（按 Ctrl+C 停止）`);

        ch.consume(NORMAL_QUEUE, (msg) => {
            if (!msg) return;

            const content = msg.content.toString();
            console.log(`[${NORMAL_QUEUE} 消费者] 收到消息: ${content}`);

            // 模拟业务逻辑判断
            const shouldAccept = content.includes('成功');
            if (shouldAccept) {
                ch.ack(msg);
                console.log(`[${NORMAL_QUEUE} 消费者] 已确认消息`);
            } else {
                ch.nack(msg, false, false);
                console.log(`[${NORMAL_QUEUE} 消费者] 已拒绝消息，进入死信队列`);
            }
        }, { noAck: false });

    } catch (error) {
        console.error(`[${NORMAL_QUEUE} 消费者] 启动失败:`, error.message);
    }
}

/**
 * 死信队列消费者
 */
async function consumeDeadQueue() {
    try {
        const ch = await getChannel();
        console.log(`[${DEAD_QUEUE} 消费者] 开始监听...（按 Ctrl+C 停止）`);

        ch.consume(DEAD_QUEUE, (msg) => {
            if (!msg) return;

            const content = msg.content.toString();
            const deathInfo = msg.properties.headers?.['x-death']?.[0] || {};
            console.log(`[${DEAD_QUEUE} 消费者] 收到死信:
        内容: ${content}
        原因: ${deathInfo.reason || '未知'}
        原始队列: ${deathInfo.queue || '未知'}
      `);
            ch.ack(msg);
        }, { noAck: false });

    } catch (error) {
        console.error(`[${DEAD_QUEUE} 消费者] 启动失败:`, error.message);
    }
}

// 优雅退出处理
async function gracefulShutdown() {
    console.log('\n开始优雅关闭...');
    if (channel && channel.close) {
        try {
            await channel.close();
            console.log('通道已关闭');
        } catch (err) {
            console.error('关闭通道出错:', err.message);
        }
    }
    if (connection && connection.close) {
        try {
            await connection.close();
            console.log('连接已关闭');
        } catch (err) {
            console.error('关闭连接出错:', err.message);
        }
    }
    process.exit(0);
}

// 监听退出信号
process.on('SIGINT', gracefulShutdown);
process.on('SIGTERM', gracefulShutdown);

// 测试流程 消费和投递消息不可以同时开启
(async () => {
    try {
        // 1. 初始化环境 始终开启这个方法
        await initRabbitMQ();

        // 2. 启动消费者（根据测试需求选择）
        // consumeNormalQueue();
        consumeDeadQueue();

        // 3. 发送测试消息
        // setTimeout(() => {
        //   produceMessage('这条消息会被正常处理（成功）', 15000);
        //   produceMessage('这条消息会被拒绝（失败）', 15000);
        // }, 1000);

    } catch (error) {
        console.error('测试流程启动失败:', error.message);
    }
})();